package com.gxlevi.db.task

import java.util

import com.alibaba.fastjson.JSON
import com.gxlevi.db.bean.{Canal, HBaseOperation}
import com.gxlevi.db.util.HBaseUtil
import org.apache.flink.streaming.api.scala.DataStream

import scala.collection.JavaConverters._
import org.apache.flink.api.scala._

import scala.collection.mutable

case class NameValuePair(
                          var columnName: String,
                          var columnValue: String,
                          var isValid: Boolean
                        )

object PreprocessTask {
  def process(canalDataStream: DataStream[Canal]) = {
    canalDataStream.flatMap {
      canal =>
        val javaList: util.List[NameValuePair] = JSON.parseArray(canal.columnValueList, classOf[NameValuePair])
        val nameValueList: mutable.Buffer[NameValuePair] = javaList.asScala
        val opType = canal.eventType
        val tableName = "mysql." + canal.dbName + "." + canal.tableName
        val cfName = "info"
        val rowkey = nameValueList(0).columnValue
        opType match {
          case "INSERT" =>
            nameValueList.map {
              nameValue => HBaseOperation(opType, tableName, cfName, rowkey, nameValue.columnName, nameValue.columnValue)
            }
          case "UPDATE" => {
            nameValueList.filter(_.isValid).map {
              nameValue => HBaseOperation(opType, tableName, cfName, rowkey, nameValue.columnName, nameValue.columnValue)
            }
          }
          case "DELETE" => {
            List(HBaseOperation(opType, tableName, cfName, rowkey, "", ""))
          }
        }
    }
  }
}
